package com.bbxdemo.kafka.produce;


import com.bbxdemo.kafka.config.MyPartitioner;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.Properties;

/**
 * produce 分区相关demo
 * kafka 默认分区相关策略
 * 1、指定分区，使用指定分区
 * 2、不指定分区，指定key ，按照key 的hashcode 对分区数 取模
 * 例： key的hashcode 为11 ，集群的分区设定为2 ， 11%2=1 ，测数据会发送到1号分区
 * 3、既不指定分区，有不指定key，kafka 采用 Sticky Partition 会随机选择一个分区，
 * 并尽可能的使用该分区，直到改分区的 batch 满或者以完成（时间到），
 * kafka 再随机选择一个与上次不同的分区
 *
 *  */
@Slf4j
@RestController
@RequestMapping("/produce/partition")
public class PartitionDemo {




    /**
     * 指定分区  异步发送消息
     * @param msg
     */
    @RequestMapping("/fixedPartition")
    public void fixedPartition(String msg,Integer partition){
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        //创建发送的内容
        ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic",partition,"",msg+"--"+System.currentTimeMillis());
        //发送数据
        producer.send(producerRecord);
        //关闭资源
        producer.close();

    }


    /**
     * 自定义分区  异步发送消息
     * @param msg
     */
    @RequestMapping("/myPartitioner")
    public void myPartitioner(String msg){
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置自定义分区
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, new MyPartitioner().getClass().getName());

        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        //创建发送的内容
        ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic",msg+"--"+System.currentTimeMillis());
        //发送数据
        producer.send(producerRecord);
        //关闭资源
        producer.close();

    }


}


